package com.wyt.mq.kafka;

import org.apache.kafka.clients.producer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.Properties;
import java.util.concurrent.Future;

/**
 * @ClassName Producer
 * @Description: TODO
 * @Author wangyongtao
 * @Date 2021/7/25 12:11
 * @Version 1.0
 **/
public class KafkaProducerDemo {
    private final static Logger logger = LoggerFactory.getLogger(KafkaProducerDemo.class);
    /**
     * 1、生产者：发送对象消息 至kafka上,调用json转化为json字符串，应为kafka存储的是String。
     * @TODO 可以自定义实现Producer接口，这样可以控制在kafka发送过程中做些自定义处理
     */
    public void sendMsgToKafka(String msg) {
        System.out.println(msg.length());
        Properties props = new Properties();
        //kafka服务器ip:端口号，集群用逗号分隔
        props.put("bootstrap.servers", "10.194.98.45:9092,10.194.98.43:9092,10.194.98.48:9092");
        // acks=0表明producer完全不管发送结果；
        // acks=all或-1表明producer会等待ISR所有节点均写入后的响应结果；
        // acks=1，表明producer会等待leader写入后的响应结果
        props.put("acks", "0");
        // 发生可重试异常时的重试次数
        props.put("retries", 0);
        // producer会将发往同一分区的多条消息封装进一个batch中，batch.size和linger.ms有一个满足就会发送消息
        // 当batch满了的时候，发送其中的所有消息,不过并不总是等待batch满了才发送消息；
        props.put("batch.size", 16384);
        // 控制消息发送延时，默认为0，即立即发送，无需关心batch是否已被填满；
        props.put("linger.ms", 50);
        // 指定了producer用于缓存消息的缓冲区大小，单位字节，默认32MB
        // producer启动时会首先创建一块内存缓冲区用于保存待发送的消息，然后由另一个专属线程负责从缓冲区中读取消息执行真正的发送
        props.put("buffer.memory", 33554432);
        // 设置producer能发送的最大消息大小
        props.put("max.request.size",10485760);
        // 设置是否压缩消息，默认none
        props.put("compression.type","lz4");
        // 设置消息发送后，等待响应的最大时间
        props.put("request.timeout.ms",30);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //因为传输格式为ByteArraySerializer，所以处理比较麻烦。如果是字符串就简单很多。
        String data = String.valueOf(new Date());
        Producer<String, String> producer = new KafkaProducer<String, String>(props) ;
        producer.send(new ProducerRecord<String, String>("WYT_KAFKA_TEMP", data, msg));
        producer.close();
    }




}